package com.lefu.risk.storm.bolt;


import com.lefu.risk.storm.common.constants.Constant;
import com.lefu.risk.storm.to.OrderDataTo;
import com.lefu.risk.storm.utils.JedisPoolUtil;
import org.apache.storm.shade.org.apache.commons.lang.StringUtils;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by mwang on 2016/12/29.
 */
public class OrderDataBolt extends BaseRichBolt {
    private OutputCollector collector;

    //普通指标(成功交易金额,失败交易金额)
    private HashMap<String,Double> commonMap = null;
    //<时间天,金额> 商户当天最大交易金额
    private HashMap<String,Double>  largestDayAmountMap = null;

    private  static long count = 0L;

    private JedisPoolUtil jedisPoolUtil = null;
    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.commonMap = new HashMap<String,Double>(){
            {
                put(Constant.COMMON_MAP_KEY_TRANS_AMOUNT_SUCCESS, 0.0D);
                put(Constant.COMMON_MAP_KEY_TRANS_AMOUNT_FAILED,0.0D);
            }
        };

        this.largestDayAmountMap = new HashMap<String,Double>();
        this.collector = collector;
        jedisPoolUtil = new JedisPoolUtil(Constant.REDIS_PROPERTIES_NAME);
    }

    @Override
    public void execute(Tuple input) {
        try{
            long time = Long.parseLong(jedisPoolUtil.time().get(0)) * Constant.SECOND_TO_MICROSECONDS + Long.parseLong(jedisPoolUtil.time().get(1));
            String[] strArr = input.getString(0).split("\t");
            if(strArr[1].equals(Constant.MATCH_TRADING_STATUS_SUCCESS) || strArr[1].equals(Constant.MATCH_TRADING_STATUS_SETTLED)){
                handleTransAmount(strArr,Constant.COMMON_MAP_KEY_TRANS_AMOUNT_SUCCESS);
            }else{
                handleTransAmount(strArr,Constant.COMMON_MAP_KEY_TRANS_AMOUNT_FAILED);
            }
            handleLargestDayAmount(strArr);
            System.err.println("执行次数----------------> " + count ++);
            OrderDataTo to = new OrderDataTo(commonMap,largestDayAmountMap);
            collector.emit(new Values(to,strArr[0],time));
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("orderDataTo","id","orderDataBoltTime"));
    }

    //处理当天最大交易
    private void handleLargestDayAmount(String[] strArr) {
        String strTime = strArr[7].split(" ")[0];
        if (!largestDayAmountMap.containsKey(strTime)){
             largestDayAmountMap.put(strTime, 0.0D);
        }
        if (StringUtils.isNotBlank(strArr[2]) && (largestDayAmountMap.get(strTime) < Double.valueOf(strArr[2]))){
            largestDayAmountMap.put(strTime,Double.valueOf(strArr[2]));
            System.err.println("strTime -------》" + strTime + "            " + "当天最大交易金额-------》" + largestDayAmountMap.get(strTime));
        }
    }


    //处理成功失败交易金额
    private void handleTransAmount(String[] strArr ,String flag) {
        double temp = commonMap.get(flag);
        if(StringUtils.isNotBlank(strArr[2])){
            commonMap.put(flag,(temp + Double.valueOf(strArr[2])));
            System.err.println("flag------>"+ flag +"   value is  ------->   " +  temp );
        }
    }
}
